Source code for hysop.backend.hardware.cpu

# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from hysop.backend.hardware.hwinfo import TopologyObject


[docs] class ProcessingUnit(TopologyObject): """ The smallest physical execution unit that hwloc recognizes. For example, there may be multiple PUs on a core (e.g., hardware threads). """ def __init__(self, parent, pu): super().__init__(parent, pu) def _parsed_type(self): return "PU" def _parse_object(self, it): raise ValueError(f"Unknown object type {_type}.")
[docs] class PhysicalCore(TopologyObject): """ A single, physical processing unit which may still contain multiple logical processors, such as hardware threads. """ def __init__(self, parent, core): self._processing_units = [] super().__init__(parent, core)
[docs] def processing_units(self): return sorted(self._processing_units, key=lambda x: x.os_index())
[docs] def processing_units_count(self): return len(self._processing_units)
def _parsed_type(self): return "Core" def _parse_object(self, it): _type = it.attrib["type"] if _type == "PU": obj = ProcessingUnit(self, it) self._processing_units.append(obj) else: raise ValueError(f"Unknown object type {_type}.") def __str__(self): return f"core {self.os_index():>2}: {self.cpu_mask()}"
[docs] class CpuPackage(TopologyObject): """ A physical package or chip, that goes into a package, it is a grouping of one or more processors. """ def __init__(self, parent, package): self._physical_cores = [] super().__init__(parent, package)
[docs] def physical_cores(self): return sorted(self._physical_cores, key=lambda x: x.os_index())
[docs] def family_number(self): return self.attribute("cpu_family_number", 0, int)
[docs] def stepping(self): return self.attribute("cpu_stepping", 0, int)
[docs] def vendor(self): return self.attribute("cpu_vendor")
[docs] def model(self): return self.attribute("cpu_model")
[docs] def physical_cores_count(self): return len(self._physical_cores)
[docs] def processing_units_count(self): return sum(x.processing_units_count() for x in self._physical_cores)
def __str__(self): header = f"::Package {self.os_index()}::" content = """ vendor: {} model: {} family: {} stepping: {} phys. cores: {} proc. units: {} cpuset: {} """.format( self.vendor(), self.model(), self.family_number(), self.stepping(), self.physical_cores_count(), self.processing_units_count(), self.cpu_mask(), ) for core in self.physical_cores(): content += str(core) + "\n" return header + self.indent(content) def _parsed_type(self): return "Package" def _parse_object(self, it): _type = it.attrib["type"] if _type == "Core": obj = PhysicalCore(self, it) self._physical_cores.append(obj) else: raise ValueError(f"Unknown object type {_type}.")